package com.gzhryc.common.logging.impls;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.lang3.StringUtils;

import com.gzhryc.common.json.JsonTools;
import com.gzhryc.common.kafka.JFKafkaProducer;
import com.gzhryc.common.kafka.models.KafkaMessage;
import com.gzhryc.common.logging.LogAnalysisUtils;
import com.gzhryc.common.logging.Logger;
import com.gzhryc.common.logging.LoggerConfigHelper;
import com.gzhryc.common.logging.enums.LogLevel;
import com.gzhryc.common.logging.models.LogInfo;

public class KafkaLogger extends Logger {

	private String tag;
	private Logger logger;

	private static BlockingQueue<LogInfoSource> queue;
	private static AtomicBoolean isRunning;
	private static Thread mainThread;

	static {
		queue = new LinkedBlockingQueue<LogInfoSource>();
		isRunning = new AtomicBoolean(false);
		mainThread = new LogMainThread();
		mainThread.start();
	}

	public KafkaLogger(DefaultLogger logger, String tag) {
		this.logger = logger;
		this.tag = tag;
	}

	@Override
	public boolean isDebug() {
		return true;
	}

	@Override
	public void debug(String message, Object... objs) {
		try {
			if (isRunning.get()) {
				LogInfoSource logInfo = new LogInfoSource(LogLevel.debug.name(), tag, message, objs);
				if (objs != null && objs.length > 0) {
					if (objs.length >= 1) {
						logInfo.setPargramType(objs[0].toString());
					}
					if (objs.length >= 2) {
						logInfo.setProjectId(objs[1].toString());
					}
					queue.add(logInfo);
					return;
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		if (logger != null) {
			logger.debug(message, objs);
		}
	}

	@Override
	public void info(String message, Object... objs) {
		try {
			if (isRunning.get()) {
				LogInfoSource logInfo = new LogInfoSource(LogLevel.info.name(), tag, message, objs);
				if (objs != null && objs.length > 0) {
					if (objs.length >= 1) {
						logInfo.setPargramType(objs[0].toString());
					}
					if (objs.length >= 2) {
						logInfo.setProjectId(objs[1].toString());
					}
					queue.add(logInfo);
					return;
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		if (logger != null) {
			logger.info(message, objs);
		}
	}

	@Override
	public void error(String message, Object... objs) {
		try {
			if (isRunning.get()) {
				LogInfoSource logInfo = new LogInfoSource(LogLevel.error.name(), tag, message, objs);
				if (objs != null && objs.length > 0) {
					if (objs.length >= 1) {
						logInfo.setPargramType(objs[0].toString());
					}
					if (objs.length >= 2) {
						logInfo.setProjectId(objs[1].toString());
					}
					queue.add(logInfo);
					return;
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		if (logger != null) {
			logger.error(message, objs);
		}
	}

	@Override
	public void warn(String message, Object... objs) {
		try {
			if (isRunning.get()) {
				LogInfoSource logInfo = new LogInfoSource(LogLevel.warn.name(), tag, message, objs);
				if (objs != null && objs.length > 0) {
					if (objs.length >= 1) {
						logInfo.setPargramType(objs[0].toString());
					}
					if (objs.length >= 2) {
						logInfo.setProjectId(objs[1].toString());
					}
					queue.add(logInfo);
					return;
				}
				queue.add(logInfo);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		if (logger != null) {
			logger.warn(message, objs);
		}
	}

	static class LogMainThread extends Thread {

		List<LogWorkThread> workThreads;

		public LogMainThread() {
			this.workThreads = new ArrayList<LogWorkThread>();
		}

		public void run() {
			isRunning.set(true);
			while (isRunning.get()) {
				while (workThreads.size() < 3) {
					workThreads.add(new LogWorkThread());
				}
				List<Integer> closeThreads = new ArrayList<Integer>();
				for (int i = 0; i < workThreads.size(); i++) {
					LogWorkThread thread = workThreads.get(i);
					if (!thread.isAlive()) {
						closeThreads.add(i);
					}
				}
				if (closeThreads.size() > 0) {
					for (int i : closeThreads) {
						workThreads.remove(i);
					}
				}
				try {
					Thread.sleep(13000L);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	}

	static class LogWorkThread extends Thread {

		final JFKafkaProducer producer;

		public LogWorkThread() {
			this.producer = LoggerConfigHelper.self().getKafkaProducer();
		}

		public void run() {
			while (isRunning.get()) {
				try {
					LogInfoSource source = queue.poll(7, TimeUnit.SECONDS);
					if (source != null && StringUtils.isNotEmpty(source.getPargramType()) && source.getMsg() != null) {
						String logMsg = LogAnalysisUtils.out(source.getMsg(), source.getObjs());
						
						LogInfo logInfo = new LogInfo(source.getLevel(), source.getTag(), logMsg);
						logInfo.setPargramType(source.getPargramType());
						logInfo.setProjectId(source.getProjectId());
						logInfo.setCreateDate(source.getCreateDate());

						StringBuffer content = new StringBuffer();
						if (source.getThrowable() != null) {
							StackTraceElement[] message = source.getThrowable().getStackTrace();
							for (StackTraceElement element : message) {
								content.append(element.toString() + "\n");
							}
						}
						logInfo.setErrorInfo(content.toString());

						KafkaMessage msg = new KafkaMessage();
						msg.setKey(source.getPargramType());
						msg.setTopic(LoggerConfigHelper.self().getKafkaTopic());
						msg.setValue(JsonTools.toJson(logInfo));
						producer.send(msg);
					}
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
	}

	static class LogInfoSource implements Serializable {

		private static final long serialVersionUID = 1L;

		private String level;
		private String tag;
		private String msg;
		private Object[] objs;
		private Throwable throwable;
		private Date createDate;

		// 扩展参数
		private String pargramType; // 程序类型
		private String projectId; // 项目ID

		public LogInfoSource(String level, String tag, String msg, Object... objs) {
			this.level = level;
			this.tag = tag;
			this.msg = msg;
			this.objs = objs;
			this.createDate = new Date();
			if (objs != null && objs.length > 0) {
				for (Object obj : objs) {
					if (Throwable.class.isInstance(obj)) {
						throwable = Throwable.class.cast(obj);
						break;
					}
				}
			}
		}

		public String getLevel() {
			return level;
		}

		public void setLevel(String level) {
			this.level = level;
		}

		public String getTag() {
			return tag;
		}

		public void setTag(String tag) {
			this.tag = tag;
		}

		public String getMsg() {
			return msg;
		}

		public void setMsg(String msg) {
			this.msg = msg;
		}

		public Throwable getThrowable() {
			return throwable;
		}

		public void setThrowable(Throwable throwable) {
			this.throwable = throwable;
		}

		public Date getCreateDate() {
			return createDate;
		}

		public void setCreateDate(Date createDate) {
			this.createDate = createDate;
		}

		public String getProjectId() {
			return projectId;
		}

		public void setProjectId(String projectId) {
			this.projectId = projectId;
		}

		public String getPargramType() {
			return pargramType;
		}

		public void setPargramType(String pargramType) {
			this.pargramType = pargramType;
		}

		public Object[] getObjs() {
			return objs;
		}

		public void setObjs(Object[] objs) {
			this.objs = objs;
		}
	}
}
